package mr.stocks.question3;

import java.awt.image.BufferedImage;
import java.io.File;
import java.io.FileOutputStream;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import mr.stocks.avro.model.StockPriceByYearReportAvro;
import mr.stocks.avro.model.StockPriceByYearValueAvro;
import mr.stocks.support.ChartBuilder;
import mr.stocks.support.StockAverage;

import org.apache.avro.mapred.AvroValue;
import org.apache.avro.mapreduce.AvroMultipleOutputs;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.jfree.chart.ChartUtilities;
import org.jfree.chart.JFreeChart;

import au.com.bytecode.opencsv.CSVReader;

public class StockPriceByYearAvroReducer
		extends
		Reducer<Text, AvroValue<StockPriceByYearReportAvro>, NullWritable, NullWritable> {
	
	public final static String SYMBOL_DESCRIPTION_FILE = "symbol_description.csv";
	private final Map<String, String> stockNameLoopup = new HashMap<String, String>();

	private AvroValue<StockPriceByYearReportAvro> outValueAvro = new AvroValue<StockPriceByYearReportAvro>();
	private StockPriceByYearReportAvro outValue = new StockPriceByYearReportAvro();
	private Map<CharSequence, StockPriceByYearValueAvro> priceMap = new HashMap<CharSequence, StockPriceByYearValueAvro>();

	private String outputPath;
	private String stockSymbol;
	private FileSystem fs;
	
	@Override
	public void setup(Context context) {
		try {
			fs = FileSystem.get(context.getConfiguration());
		} catch (IOException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		}
		
		outputPath = context.getConfiguration().get("report.output.path");
	
		FileReader reader = null;
		CSVReader csvReader = null;
		try {
			reader = new FileReader(SYMBOL_DESCRIPTION_FILE);
			csvReader = new CSVReader(reader);
			List<String[]> entries = csvReader.readAll();
			Iterator<String[]> i = entries.iterator();

			while (i.hasNext()) {
				String[] tokens = i.next();
				stockNameLoopup.put(tokens[0], tokens[1]);
			}
		}catch (IOException e){
			e.printStackTrace();
		}finally {
			IOUtils.closeQuietly(csvReader);
			IOUtils.closeQuietly(reader);
		}
	}

	@Override
	protected void cleanup(Context context) throws IOException,
			InterruptedException {
	}

	@Override
	protected void reduce(Text inKey,
			Iterable<AvroValue<StockPriceByYearReportAvro>> inValues,
			Context context) throws IOException, InterruptedException {

		priceMap.clear();
    	
    	for(AvroValue<StockPriceByYearReportAvro> value : inValues){
    		
    		Map<CharSequence, StockPriceByYearValueAvro> inMap = value.datum().getAverageAnnualPrice();
    		for(CharSequence key: inMap.keySet()){
    			if(priceMap.containsKey(key)){
    				StockPriceByYearValueAvro stockAnnaulValue = 
    						priceMap.get(key);
    				
    				stockAnnaulValue.setCount(stockAnnaulValue.getCount() + inMap.get(key).getCount());
    				stockAnnaulValue.setPrice(stockAnnaulValue.getPrice() + inMap.get(key).getPrice());
    			}
    			else
    				priceMap.put(key, inMap.get(key));
    		}
    	}
    	
    	stockSymbol = inKey.toString();
    	outValue.setStockSymbol(stockSymbol);
    	outValue.setAverageAnnualPrice(priceMap);
    	outValueAvro.datum(outValue);
    	
    	//create chart and save PNG to disk
    	FSDataOutputStream outStream = null;
    	Path outputFile = null;
		try {
			outputFile = new Path(outputPath + "/" + stockSymbol + ".png");
			outStream = fs.create(outputFile);
			List<StockAverage> data = new ArrayList<StockAverage>();

			for (CharSequence key : outValue.getAverageAnnualPrice().keySet()) {
				double price = outValue.getAverageAnnualPrice().get(key)
						.getPrice()/ outValue.getAverageAnnualPrice().get(key).getCount();
				data.add(new StockAverage(stockSymbol, key.toString(),
						(float) price));
			}

			String stockName = null;
			if ((stockName = stockNameLoopup.get(stockSymbol)) != null)
				stockName += "(" + this.stockSymbol + ")";
			else
				stockName = this.stockSymbol;

			ChartBuilder chartBuilder = new ChartBuilder(stockName, "Year", "Dividents", data);
			final JFreeChart chart = chartBuilder.build();
			BufferedImage img = chart.createBufferedImage(1200, 600);
			ChartUtilities.writeBufferedImageAsPNG(outStream, img);
		} finally {
			if(outStream != null)
				outStream.close();
		}
	
	}

}
